package com.zhang.third.day06;

import com.zhang.third.utils.ClickEventSource;
import com.zhang.third.utils.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;

/**
 * @title: 查询流
 * @author: zhang
 * @date: 2022/4/9 15:50
 */
public class Example8 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Event> clickSource = env.addSource(new ClickEventSource());

        DataStreamSource<String> queryStream = env.socketTextStream("localhost", 9999);

        clickSource
                .keyBy(r->r.key)
                .connect(queryStream.setParallelism(1).broadcast())
                .flatMap(new Query())
                .print();

        env.execute();
    }

    public static class Query implements CoFlatMapFunction<Event,String,Event>{
        private String queryStream = "";
        @Override
        public void flatMap1(Event in, Collector<Event> out) throws Exception {
            if (queryStream.equals(in.value)) out.collect(in);
        }

        @Override
        public void flatMap2(String value, Collector<Event> out) throws Exception {
            queryStream = value;
        }
    }
}
